
#include <stdio.h>
#include <sys/file.h>
#include <string.h>
#include <sstream>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include "bocat/common/base64.hpp"

#include "boque/buffesq.hpp"


namespace   boque   {


///////////////////////////////////////////////////////////////////////
//

class	GuardByFileLock
{
public	:
	GuardByFileLock( int fd )
		:	_fd( fd )
	{
		flock( _fd, LOCK_EX );
	}

	~GuardByFileLock()
	{
		flock( _fd, LOCK_UN );
	}

private	:
	GuardByFileLock( const GuardByFileLock& );
	GuardByFileLock&	operator=( const GuardByFileLock& );

private	:
	int	_fd;
};


///////////////////////////////////////////////////////////////////////
//

#define STATUS_FILE "status"

#define DEFAULT_BUF_SIZE    (1024 * 1024 * 4)

#define FILE_SIZE_THRESHOLD (1024 * 1024 * 256)


BuffesQ::BuffesQ( const std::string& q_dir, u_int32_t file_size_threshold )
    :   _q_dir( q_dir ),
        _fp_status( NULL ),
        _fp_push( NULL ), _fp_pop( NULL )
{
    boost::filesystem::path q_path( _q_dir );
    if ( !boost::filesystem::exists( q_path ) ) {
        if ( !boost::filesystem::create_directories( q_path ) ) {
            std::ostringstream  msg;
            msg << "fail to create buffes-q directory " << _q_dir;

            throw std::runtime_error( msg.str() );
        }
    }

    _file_size_threshold = file_size_threshold;
    if ( _file_size_threshold < 1024 ) {
        _file_size_threshold = FILE_SIZE_THRESHOLD;
    }

    _status_file = _q_dir + "/" + STATUS_FILE;

    openStatusFile();
    openPushFile();
    openPopFile();

    _buf_size = DEFAULT_BUF_SIZE;
    _buf_p = (char *)malloc( _buf_size );
    if ( NULL == _buf_p ) {
        std::ostringstream  msg;
        msg << "fail to malloc buffer (size=" << _buf_size << ")";

        throw std::runtime_error( msg.str() );
    }
}

BuffesQ::~BuffesQ()
{
    closePopFile();
    closePushFile();
    closeStatusFile();
}

///////////////////////////////////////////////////////////////////////
//

#define DATA_MAGIC_STRING   "BuffesQ.DATA   \n"
#define DATA_MAGIC_STRING_SIZE  16

#define STATUS_MAGIC_STRING "BuffesQ.STATUS \n"
#define STATUS_MAGIC_STRING_SIZE    16

void
BuffesQ::push( const std::string& data )
{
	push( data.c_str(), data.size() );
}

void
BuffesQ::push( const char* data, size_t size )
{
	int	fd_status = getStatusFD();
	GuardByFileLock	guard( fd_status );

    boost::mutex::scoped_lock   lock( _mutex_push );

    fseek( _fp_push, 0, SEEK_END );
    
    std::string data_b64 = bocat::common::base64_encode( data, size );
    size_t rc = fwrite( data_b64.c_str(), data_b64.size(), 1, _fp_push );
	if ( rc < 1 ) {
		std::ostringstream  msg;
		msg << "fail to push (fileno=" << _tail.fileno << ")";
		throw std::runtime_error( msg.str() );
	}
    rc = fwrite( "\n", 1, 1, _fp_push );
	if ( rc < 1 ) {
		std::ostringstream  msg;
		msg << "fail to push (fileno=" << _tail.fileno << ")";
		throw std::runtime_error( msg.str() );
	}
    long    offset = ftell( _fp_push );
    if ( offset >= _file_size_threshold ) {
        //switch Push File;
        closePushFile();

        _tail.fileno = getNextDataFileNo( _tail.fileno );
        _tail.offset = DATA_MAGIC_STRING_SIZE;

        updateStatusTailLocation();

        openPushFile();
    }
    else {
        _tail.offset = offset;
        updateStatusTailLocation();
    }
}

std::string
BuffesQ::front( void )
{
	int	fd_status = getStatusFD();
	GuardByFileLock	guard( fd_status );

    boost::mutex::scoped_lock   lock( _mutex_pop );

    if ( empty() ) {
        return  std::string();
    }

	return	_front();
}


std::string
BuffesQ::pop( void )
{
	int	fd_status = getStatusFD();
	GuardByFileLock	guard( fd_status );

    boost::mutex::scoped_lock   lock( _mutex_pop );

    if ( empty() ) {
        return  std::string();
    }

	std::string	data = _front();

	long    offset = ftell( _fp_pop );
	_head.offset = offset;
	updateStatusHeadLocation();

	return  data;
}

std::string
BuffesQ::_front( void )
{
    while ( true ) {
        fseek( _fp_pop, _head.offset, SEEK_SET );
        
        *( _buf_p + _buf_size - 1 ) = '\0';
        *( _buf_p + _buf_size - 2 ) = '\0';
        char*   s = fgets( _buf_p, _buf_size, _fp_pop );
        if ( NULL == s ) {
            if ( _head.offset < _file_size_threshold ) {
                std::ostringstream  msg;
                msg << "fail to pop (fileno=" << _head.fileno << ", offset=" << _head.offset << ")";
                throw std::runtime_error( msg.str() );
            }

            // open next file
            closePopFile();

            removeDataFile( _head.fileno );

            _head.fileno = getNextDataFileNo( _head.fileno );
            _head.offset = DATA_MAGIC_STRING_SIZE;
            updateStatusHeadLocation();

            openPopFile();

            continue;
        }

        if ( ('\0' != *(_buf_p + _buf_size - 2)) && ('\n' != *(_buf_p + _buf_size - 2)) ) {
            enlargeBuffer();
            continue;
        }

        // got it
		// skip the last '\n' before base64::decode
        std::string data_b64( _buf_p, strlen(_buf_p) - 1 );
        boost::trim( data_b64 );
        std::string data = bocat::common::base64_decode< std::string >( data_b64 );

        return  data;
    }
}


bool
BuffesQ::empty( void )
{
    if ( _head != _tail ) {
        return  false;
    }
    if ( _head == _tail ) {
        reloadStatusLocation();
        if ( _head != _tail ) {
            return  false;
        }
    }

    return  true;
}

///////////////////////////////////////////////////////////////////////
//


///////////////////////////////////////////////////////////////////////
//

size_t  
BuffesQ::getStatusHeadOffset( void )
{
    return  DATA_MAGIC_STRING_SIZE;
}

size_t
BuffesQ::getStatusTailOffset( void )
{
    return  DATA_MAGIC_STRING_SIZE + sizeof(Location);
}

void
BuffesQ::updateStatusHeadLocation( void )
{
    boost::mutex::scoped_lock   lock( _mutex_status );

    if ( NULL == _fp_status ) {
        openStatusFile();
    }

    fseek( _fp_status, getStatusHeadOffset(), SEEK_SET );
    fwrite( &_head, sizeof(Location), 1, _fp_status );
}

void
BuffesQ::updateStatusTailLocation( void )
{
    boost::mutex::scoped_lock   lock( _mutex_status );

    if ( NULL == _fp_status ) {
        openStatusFile();
    }

    fseek( _fp_status, getStatusTailOffset(), SEEK_SET );
    fwrite( &_tail, sizeof(Location), 1, _fp_status );
}

void
BuffesQ::reloadStatusLocation( void )
{
    boost::mutex::scoped_lock   lock( _mutex_status );

    fseek( _fp_status, getStatusHeadOffset(), SEEK_SET );
    size_t  head_size = fread( &_head, 1, sizeof(Location), _fp_status );
    if ( head_size != sizeof(Location) ) {
        std::ostringstream  msg;
        msg << "fail to read HEAD location from status file " << _status_file;
        throw std::runtime_error( msg.str() );
    }

    fseek( _fp_status, getStatusTailOffset(), SEEK_SET );
    size_t  tail_size = fread( &_tail, 1, sizeof(Location), _fp_status );
    if ( tail_size != sizeof(Location) ) {
        std::ostringstream  msg;
        msg << "fail to read TAIL location from status file " << _status_file;
        throw std::runtime_error( msg.str() );
    }
}


void    
BuffesQ::openStatusFile( void )
{
    boost::filesystem::path status_path( _status_file );
    if ( !boost::filesystem::exists( status_path ) ) {
        _fp_status = fopen( _status_file.c_str(), "w" );
        fwrite( STATUS_MAGIC_STRING, STATUS_MAGIC_STRING_SIZE, 1, _fp_status );
        Location    loc( 1, DATA_MAGIC_STRING_SIZE );
        // write head location
        fwrite( &loc, sizeof(Location), 1, _fp_status );
        // write tail location
        fwrite( &loc, sizeof(Location), 1, _fp_status );
        fclose( _fp_status );
        _fp_status = NULL;
    }

    _fp_status = fopen( _status_file.c_str(), "r+" );
    setbuf( _fp_status, NULL );

    reloadStatusLocation();
}

void    
BuffesQ::openPushFile( void )
{
    char    buf[ 64 ];
    memset( buf, 0, sizeof(buf) );
    snprintf( buf, sizeof(buf) - 1, "q.%09d", _tail.fileno );
    std::string push_file = _q_dir + "/" + buf;
    createDataFile( push_file );

    _fp_push = fopen( push_file.c_str(), "a" );
    setlinebuf( _fp_push );
    fseek( _fp_push, 0, SEEK_END );

    int32_t pos = ftell( _fp_push );
    if ( pos != _tail.offset ) {
        std::ostringstream  msg;
        msg << "Bad offset of PUSH file " << push_file << " (GOT " << pos << ", WANT " << _tail.offset << ")";
        throw std::runtime_error( msg.str() );
    }
}

void    
BuffesQ::openPopFile( void )
{
    char    buf[ 64 ];
    memset( buf, 0, sizeof(buf) );
    snprintf( buf, sizeof(buf) - 1, "q.%09d", _head.fileno );
    std::string pop_file = _q_dir + "/" + buf;
    createDataFile( pop_file );

    _fp_pop = fopen( pop_file.c_str(), "r" );
    fseek( _fp_pop, _head.offset, SEEK_SET );

    int32_t pos = ftell( _fp_pop );
    if ( pos != _head.offset ) {
        std::ostringstream  msg;
        msg << "Bad offset of POP file " << pop_file << " (GOT " << pos << ", WANT " << _head.offset << ")";
        throw std::runtime_error( msg.str() );
    }
}

void    
BuffesQ::closePopFile( void )
{
    if ( NULL != _fp_pop ) {
        fclose( _fp_pop );
        _fp_pop = NULL;
    }
}

void    
BuffesQ::closePushFile( void )
{
    if ( NULL != _fp_push ) {
        fclose( _fp_push );
        _fp_push = NULL;
    }
}

void    
BuffesQ::closeStatusFile( void )
{
    if ( NULL != _fp_status ) {
        fclose( _fp_status );
        _fp_status = NULL;
    }
}

int
BuffesQ::getStatusFD( void )
{
	if ( NULL == _fp_status ) {
		openStatusFile();
	}

	return	fileno( _fp_status );
}


///////////////////////////////////////////////////////////////////////
//


void
BuffesQ::createDataFile( const std::string& filename )
{
    boost::filesystem::path file_path( filename );
    if ( !boost::filesystem::exists( file_path ) ) {
        FILE*   fp = fopen( filename.c_str(), "w" );
        fwrite( DATA_MAGIC_STRING, DATA_MAGIC_STRING_SIZE, 1, fp );
        fclose( fp );
        fp = NULL;
    }
}

void
BuffesQ::removeDataFile( int32_t file_id )
{
    char    buf[ 64 ];
    memset( buf, 0, sizeof(buf) );
    snprintf( buf, sizeof(buf) - 1, "q.%09d", file_id );
    std::string file_path = _q_dir + "/" + buf;

    if ( boost::filesystem::exists( file_path ) ) {
        boost::filesystem::remove( file_path );
    }
}


int32_t
BuffesQ::getNextDataFileNo( int32_t fileno )
{
    int32_t fileno_next = fileno + 1;
    if ( fileno_next >= 1000000000 ) {
        fileno_next = 1000;
    }

    return  fileno_next;
}

void
BuffesQ::enlargeBuffer( void )
{
    size_t  new_size = (size_t)(_buf_size * 1.5);
    void*   p = realloc( _buf_p, new_size );
    if ( NULL != p ) {
        _buf_p = (char *)p;
        _buf_size = new_size;
    }
}


}   //// namespace boque


